package org.example;



import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.alibaba.fastjson2.TypeReference;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import com.typesafe.config.ConfigList;
import com.typesafe.config.ConfigValue;
import org.example.datasource.model.FieldMetadata;
import org.example.datasource.model.TableMetadata;
import org.example.dolph.DolphinschedulerClient;
import org.example.dolph.model.PageInfo;
import org.example.dolph.model.ProjectInfo;
import org.example.dolph.model.TaskDefine;
import org.example.dolph.model.TaskDetail;
import org.example.model.ColumnLineage;
import org.example.model.TableLineage;
import org.example.seatunnel.SeatunnelSinkWrapper;
import org.example.seatunnel.SeatunnelSourceWrapper;
import org.example.seatunnel.model.SinkResModel;
import org.example.seatunnel.model.SourceResModel;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;

public class SeatunnelParser {


    public List<TableLineage> parse(String sql) {
        return null;
    }

    public static void main(String[] args) {


        DataHubClient dataHubClient = new DataHubClient("http://192.168.0.13:8080");

        SeatunnelParser seatunnelParser = new SeatunnelParser();

       // seatunnelParser.parse();

        String token = "dc6b6ef2ba096208c49b3c148a9c30d4";	// dolphin scheduler token
        String dolphinAddress = "http://192.168.0.5:32777/dolphinscheduler";  // dolphin scheduler address


        DolphinschedulerClient dolphinschedulerClient = new DolphinschedulerClient(token, dolphinAddress);

        for (ProjectInfo projectInfo : dolphinschedulerClient.getProjects().getTotalList()) {
            System.out.println(projectInfo.getCode());

            for (TaskDefine taskDefine : dolphinschedulerClient.getTaskDefines(projectInfo.getCode()).getTotalList()) {
//                System.out.println(taskDefine.getTaskName() +":" + taskDefine.getTaskType());
                if(taskDefine.getTaskType().equals("SEATUNNEL")){
                    TaskDetail taskDefine1 =  dolphinschedulerClient.getTaskDetail(projectInfo.getCode(),taskDefine.getTaskCode());
//                    System.out.println(taskDefine1.getSeaTaskParams().getRawScript());

                    Config config =  ConfigFactory.parseString(taskDefine1.getSeaTaskParams().getRawScript());

                    System.out.println(config);
                    Config configValues =  config.getConfig("source");

                    try{
                        SourceResModel sourceResModel =  SeatunnelSourceWrapper.parse(configValues);
                        if (sourceResModel != null){
                            Config sink =  config.getConfig("sink");
                            SinkResModel sinkRes = SeatunnelSinkWrapper.parse(sourceResModel,sink);


                            Map<String,List<String>> metadata = new HashMap<>();
                            for (TableMetadata tableMetadata : sinkRes.getSourceMetadataList()) {
                                List<String> fields = tableMetadata.getFields().stream().map(FieldMetadata::getFileName).collect(Collectors.toList());
                                metadata.put("source."+tableMetadata.getTableName(),fields);
                            }
                            JSONObject jsonObject = new JSONObject();
                            jsonObject.put("metadata",metadata);
                            jsonObject.put("sql",sinkRes.getSql());
                            String res = HttpUtil.post("http://127.0.0.1:8000/api/lineage",jsonObject.toJSONString());
                            Map<String,TableLineage> tableLineageMap = new HashMap<>();

                            List<List<JSONObject>> resList = JSON.parseObject(res,new TypeReference<List<List<JSONObject>>>() {});
                            for (List<JSONObject> strings : resList) {
                                JSONObject sourceTable = strings.get(0).getJSONObject("_parent");
                                String sourceTableSchema = sourceTable.getJSONObject("schema").getString("raw_name");
                                String sourceTableName = sourceTable.getString("raw_name");
                                String sourceField = strings.get(0).getString("raw_name");


                                JSONObject targetTable = strings.get(1).getJSONObject("_parent");
                                String  targetTableSchema = targetTable.getJSONObject("schema").getString("raw_name");
                                String  targetTableName = targetTable.getString("raw_name");
                                String  targetField = strings.get(1).getString("raw_name");

                                TableLineage tableLineage =  tableLineageMap.computeIfAbsent(targetTableSchema + "." + targetTableName,s->{
                                    TableLineage tableLineage1 = new TableLineage();
                                    tableLineage1.setDatabaseName(targetTableSchema);
                                    tableLineage1.setTableName(targetTableName);
                                    return tableLineage1;
                                });


                                List<ColumnLineage> columnLineages =
                                        tableLineage.getColumnLineages().computeIfAbsent(targetField,s -> new ArrayList<>());

                                ColumnLineage columnLineage = new ColumnLineage();
                                columnLineage.setSourceColumn(sourceField);
                                TableLineage sourceTableInfo = new TableLineage();
                                sourceTableInfo.setDatabaseName(targetTableSchema);
                                sourceTableInfo.setTableName(sourceTableName);
                                columnLineage.setSourceTable(sourceTableInfo);
                                columnLineages.add(columnLineage);
                            }

                            dataHubClient.sub(new ArrayList<>(tableLineageMap.values()));

                        }
                    }catch (Exception e){
                        e.printStackTrace();
                    }

                }
            }
        }



    }
}
